package com.zili.tableapiandsql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.e;

/**
 * @author : ranzlupup
 * @since : 2023/3/16 10:21
 */
public class CreateATableEnvironment {
    public static void main(String[] args) throws Exception {
        // EnvironmentSettings settings1 = EnvironmentSettings
        //         .newInstance()
        //         .inStreamingMode() // 使用流处理模式
        //         .build();
        // TableEnvironment tableEnvironment1 = TableEnvironment.create(settings1);

        // 创建表环境  使用流处理模式
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 创建表
        String createInputDDL =
                "CREATE TABLE clickTable (" +
                        " user_name STRING, " +
                        " url STRING, " +
                        " ts BIGINT " +
                        ")  WITH (" +
                        " 'connector' = 'filesystem', " +
                        " 'path' = 'input/clicks.csv'," +
                        " 'format' = 'csv' " +
                        ")";
        tableEnv.executeSql(createInputDDL);

        // 表的查询转换
        Table clickTable = tableEnv.from("clickTable");
        Table resultTable = clickTable.where($("user_name").isEqual("Bob"))
                .select($("user_name"), $("url"));
        // tableEnv.toDataStream(resultTable).print("resultTable: ");

        // 注册临时视图
        tableEnv.createTemporaryView("resultTable", resultTable);

        // 执行SQL进行表的查询转换
        Table resultTable2 = tableEnv.sqlQuery("select * from resultTable");
        // tableEnv.toChangelogStream(resultTable2).print("resultTable2: ");

        // 执行聚合计算的查询转换
        Table aggResultTable = tableEnv.sqlQuery("select user_name, COUNT(url) as cnt from resultTable group by user_name");
        // tableEnv.toChangelogStream(aggResultTable).print("aggResultTable: ");

        // 创建一张用于输出的表
        String createPrintOutDDL =
                "CREATE TABLE printOutTable (" +
                        " user_name STRING," +
                        " cnt BIGINT " +
                        ") WITH (" +
                        " 'connector' = 'print' " +
                        ")";
        tableEnv.executeSql(createPrintOutDDL);

        // 输出表
        aggResultTable.executeInsert("printOutTable");

        // env.execute();
    }
}
